package org.jetlinks.pro.things.impl.connector;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.hswebframework.web.id.IDGenerator;
import org.hswebframework.web.utils.DigestUtils;
import org.jetlinks.core.event.EventBus;
import org.jetlinks.core.event.Subscription;
import org.jetlinks.core.message.HeaderKey;
import org.jetlinks.core.message.MessageType;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.message.event.ThingEventMessage;
import org.jetlinks.core.message.property.DefaultReportPropertyMessage;
import org.jetlinks.core.message.property.PropertyMessage;
import org.jetlinks.core.message.property.ReportPropertyMessage;
import org.jetlinks.core.message.property.ThingReportPropertyMessage;
import org.jetlinks.core.things.MetadataId;
import org.jetlinks.core.things.ThingId;
import org.jetlinks.core.things.ThingMetadataType;
import org.jetlinks.pro.things.ThingConstants;
import org.jetlinks.pro.things.ThingsConnectorManager;
import org.jetlinks.pro.things.utils.ThingMetadataHelper;
import org.jetlinks.pro.things.utils.ThingsUtils;
import reactor.core.Disposable;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.util.*;
import java.util.function.Function;

/**
 * 物属性连接器,将指定的物的属性连接为新的物属性.
 *
 * @author zhouhao
 * @since 1.12
 */
@Slf4j
@AllArgsConstructor
class ThingPropertiesMessageConnector extends AbstractThingsConnector {

    /**
     * 物类型
     */
    @Nonnull
    private final String thingType;

    /**
     * 物模版(设备产品)ID
     */
    @Nonnull
    private final String templateId;

    /**
     * 物类型,不为null时,则认为从指定的物连接属性
     */
    private final String thingId;

    /**
     * 事件总线
     */
    private final EventBus eventBus;

    /**
     * 要订阅的属性
     */
    private final Set<String> properties;

    /**
     * 物连接器管理器,用于管理连接器,并从连接器中获取数据进行持久化等处理.
     */
    private final ThingsConnectorManager connectorManager;

    //连接到指定物的物模型
    @Override
    public Disposable doConnectTo(ThingId thingId,
                                  MetadataId metadataId,
                                  Function<ThingMessage, Mono<Void>> handler) {

        Function<ThingMessage, Mono<ThingMessage>> converter = createConverter(thingId, metadataId);
        String subId = this.thingId == null ?
            thingId.getId() //当前物
            : this.thingId; //其他物
        log.debug("connect thing [{}@{}] properties {} to {}", subId, thingType, properties, thingId.getId());
        return doConnect(subId, metadataId, converter, handler);
    }

    //连接到当前物的物模型
    @Override
    public Disposable doConnectTo(MetadataId metadataId,
                                  Function<ThingMessage, Mono<Void>> handler) {
        Function<ThingMessage, Mono<ThingMessage>> converter = createConverter(null, metadataId);
        String subId = this.thingId == null ? "*" : this.thingId;
        log.debug("connect things [{}/{}@{}] properties {}", templateId, subId, thingType, properties);
        return doConnect(subId, metadataId, converter, handler);
    }

    /**
     * 做消息连接
     *
     * @param thingId 物ID
     * @param metadataId 物模型ID
     * @param converter 消息转换器
     * @param handler 消息处理器
     */
    private Disposable doConnect(String thingId,
                                 MetadataId metadataId,
                                 Function<ThingMessage, Mono<ThingMessage>> converter,
                                 Function<ThingMessage, Mono<Void>> handler) {
        //创建连接器唯一标识，用于判断连接器是否递归
        final String id = IDGenerator.MD5.generate();

        //订阅属性上报和属性回复消息
        return eventBus
            .subscribe(Subscription
                           .builder()
                           .subscriberId("thing-properties-connector:" + thingId + "@" + this.thingType)
                           .topics(
                               "/" + this.thingType + "/" + this.templateId + "/" + thingId + "/message/property/report",
                               "/" + this.thingType + "/" + this.templateId + "/" + thingId + "/message/property/*/reply"
                           )
                           //只订阅本地的数据
                           .justLocal()
                           .build(),
                       ThingMessage.class)
            //过滤掉递归的消息
            .flatMap(msg -> {
                if (msg
                    .getHeaderOrElse(ThingConstants.connectorTrace, Collections::emptySet)
                    .contains(id)) {
                    log.debug("thing [{}@{}] connector [{}] has recursive", msg.getThingId(), msg.getThingType(), metadataId);
                    return Mono.empty();
                }
                //消息转换处理，通常用于将消息转换为属性上报或事件上报
                return converter
                    .apply(msg)
                    .doOnNext(_msg -> {
                        //添加追踪
                        ThingsUtils.mergeHeader(msg, _msg);
                        _msg.getOrAddHeaderDefault(ThingConstants.connectorTrace).add(id);
                    })
                    //处理消息
                    .flatMap(handler)
                    .onErrorResume((err) -> {
                        log.error(err.getMessage(), err);
                        return Mono.empty();
                    });
            })
            .subscribe();
    }


    /**
     * 根据物模型ID创建对应的消息转换器
     *
     * @param thingId 物ID
     * @param metadataId 物模型ID
     * @return 消息转换器
     */
    private Function<ThingMessage, Mono<ThingMessage>> createConverter(@Nullable ThingId thingId,
                                                                       @Nonnull MetadataId metadataId) {
        Function<ThingMessage, Mono<ThingMessage>> converter = ThingMetadataHelper
            .<ThingMessage, Mono<ThingMessage>>create()
            //连接到属性
            .when(ThingMetadataType.property, (tid, mid, msg) -> {
                //连接到属性时
                Map<String, Object> properties = convertProperties(mid.getId(), msg);
                if (MapUtils.isEmpty(properties)) {
                    return Mono.empty();
                }
                ThingMessage message = MessageType.REPORT_PROPERTY
                    //添加消息的物ID，当传入物ID为空时，使用当前消息的物ID（连接到当前物模型时，物ID通常为空）
                    .<ThingReportPropertyMessage>forThing(tid == null ? ThingId.of(msg.getThingType(), msg.getThingId()) : tid)
                    .success(properties)
                    .timestamp(msg.getTimestamp());

                return Mono.just(message);
            })
            //连接到事件
            .when(ThingMetadataType.event, (tid, mid, msg) -> {
                //事件时返回全部属性
                Map<String, Object> properties = ((PropertyMessage) msg).getProperties();
                if (MapUtils.isEmpty(properties)) {
                    return Mono.empty();
                }
                ThingMessage message = MessageType.EVENT
                    //添加消息的物ID，当传入物ID为空时，使用当前消息的物ID（连接到当前物模型时，物ID通常为空）
                    .<ThingEventMessage>forThing(tid == null ? ThingId.of(msg.getThingType(), msg.getThingId()) : tid)
                    .event(mid.getId())
                    .data(properties)
                    .timestamp(msg.getTimestamp());
                return Mono.just(message);
            })
            .toFunction(thingId, metadataId);

        if (thingId == null) {
            //没有指定物ID,则在收到数据判断物实例是否已经存在独立的Connector
            return msg -> {
                ThingId id = ThingId.of(msg.getThingType(), msg.getThingId());
                return connectorManager
                    .getConnector(id, metadataId)
                    //如果有独立的连接器,则忽略
                    .map(ignore -> Mono.<ThingMessage>empty())
                    .defaultIfEmpty(Mono.defer(() -> converter.apply(msg)))
                    .flatMap(Function.identity());
            };
        }
        return converter;
    }

    private Map<String, Object> convertProperties(String property, ThingMessage msg) {
        PropertyMessage propertyMessage = ((PropertyMessage) msg);
        //没有指定固定的属性或者设置全部属性时,则返回全部属性值
        if (CollectionUtils.isEmpty(this.properties) || this.properties.contains("*")) {
            return propertyMessage.getProperties();
        }
        Map<String, Object> properties = new HashMap<>();
        //从指定的属性里获取任意属性值
        for (String fromProperty : this.properties) {
            @SuppressWarnings("all")
            Object value = propertyMessage.getProperty(fromProperty).orElse(null);
            if (null != value) {
                properties.put(property, value);
                break;
            }
        }
        return properties;
    }

    @Override
    protected void onDispose() {

    }
}
